package temp.casia.isiteam.zrb.hbase.client;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import temp.casia.isiteam.zrb.hbase.client.bean.HBInvestorCapital;
import temp.casia.isiteam.zrb.hbase.client.bean.HBInvestorFinance;
import temp.casia.isiteam.zrb.hbase.client.bean.HBInvestorInfo;
import temp.casia.isiteam.zrb.hbase.client.bean.HBInvestorPrincipal;
import temp.casia.isiteam.zrb.hbase.client.bean.HBInvestorProduct;
import temp.casia.isiteam.zrb.hbase.client.bean.HBInvestorSocialRelations;
import temp.casia.isiteam.zrb.hbase.client.bean.HbaseStatus;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableCityInfo;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableInvestorCapital;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableInvestorFinance;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableInvestorInfo;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableInvestorPrincipal;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableInvestorProduct;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableInvestorSocialRelations;
import temp.casia.isiteam.zrb.hbase.client.util.Commons;
import temp.casia.isiteam.zrb.hbase.client.util.ConnectionManager;
import temp.casia.isiteam.zrb.hbase.client.util.MD5Util;

/**
 * 投资人信息操作类<br>
 * 提供HBase中投资人相关表的查询、写入
 * @author 彭鑫
 * @createDate 2015年6月29日
 */
public class InvestorHbaseClient {
	protected Configuration config;
	private Logger logger = Logger.getLogger(InvestorHbaseClient.class);
	/**
	 * 表名
	 */
	protected Map<String, TableName> tableNameMap;
	protected String quorum;
	protected Connection conn;
	protected Integer currentClientNum = 0;
	private Timer timer;

	/**
	 * 使用默认的master地址
	 */
	public InvestorHbaseClient(){
		this.quorum = Commons.quorum;
		this.init();
	}
	
	/**
	 * 指定master地址
	 * @param quorum
	 */
	public InvestorHbaseClient(String quorum){
		this.quorum = quorum;
		this.init();
	}
	
	protected void init() {
		try {
			config = HBaseConfiguration.create();
			config.set("hbase.zookeeper.quorum", this.quorum);
			config.set("hbase.client.retries.number", "3");
			config.set("zookeeper.recovery.retry", "3");
			
			conn = ConnectionManager.getConn(config);
			Admin admin = conn.getAdmin();
			TableName[] tableAry = admin.listTableNames("investor_.*");
			this.tableNameMap = new HashMap<String, TableName>();
			for (TableName tableName : tableAry) {
				this.tableNameMap.put(tableName.getNameAsString(), tableName);
			}
		} catch (Exception e) {
			this.logger.error("", e);
		}
	}
	
	protected void initConn() {
		this.conn = ConnectionManager.getConn(config);
	}
	
	/**
	 * 批量查询投资人基本信息
	 * @param investorIDList
	 * @return
	 */
	public List<HBInvestorInfo> getBasicInfo(List<Integer> investorIDList){
		List<HBInvestorInfo> list = new ArrayList<HBInvestorInfo>();
		this.currentClientNum++;
		Table table = null;
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap
					.get(HbaseTableCityInfo.TABLE_NAME));
			List<Get> getList = new ArrayList<Get>();
			for(Integer investorID: investorIDList){
				Get get = new Get(Bytes.toBytes(Integer.toString(investorID)));
				getList.add(get);
			}
			Result[] results = table.get(getList);
			for(Result result: results){
				HBInvestorInfo investor = this.getInvestorBean(result);
				list.add(investor);
			}
		} catch (Exception e) {
			this.logger.error("", e);
		} finally {
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		
		return list;
	}
	/**
	 * 查询投资人信息
	 * @param investorID
	 * @param industryID
	 * @return
	 */
	public HBInvestorInfo getBasicInfo(int investorID, int industryID){
		HBInvestorInfo investor = null;
		this.currentClientNum++;
		Table table = null;
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorInfo.TABLE_NAME));
			String rowKey = investorID + "";
			Get get = new Get(Bytes.toBytes(rowKey));
			Result result = table.get(get);
			if(result != null){
				investor = this.getInvestorBean(result);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return investor;
	}
	
	private HBInvestorInfo getInvestorBean(Result result){
		HBInvestorInfo investor = new HBInvestorInfo();
		String rowKey = Bytes.toString(result.getRow());
		investor.setInvestorID(Integer.parseInt(rowKey));
		investor.setName(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_name));
		investor.setIndustry(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_industry));
		investor.setProvince(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_province));
		investor.setCity(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_city));
		investor.setCounty(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_county));
		investor.setAbbreviation(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_abbreviation));
		investor.setEnName(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_enName));
		investor.setParentCompany(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_parentCompany));
		investor.setRegisteredNo(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_registeredNo));
		investor.setLegalPerson(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_legalPerson));
		investor.setRegDate(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_regDate));
		investor.setDateRange(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_dateRange));
		investor.setOrgCode(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_orgCode));
		investor.setIsIPO(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_isIPO));
		investor.setStockCode(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_stockCode));
		investor.setProperty(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_property));
		investor.setRegCapital(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_regCapital));
		investor.setAddress(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_address));
		investor.setWebsite(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_website));
		investor.setWeixin(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_weixin));
		investor.setWeibo(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_weibo));
		investor.setTelephone(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_telephone));
		investor.setEmail(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_email));
		investor.setFax(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_fax));
		investor.setBusinessScope(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_businessScope));
		investor.setContackAdd(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_contackAdd));
		investor.setOrgForm(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_orgForm));
		investor.setOfficeAdd(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_officeAdd));
		investor.setEmployees(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_employees));
		investor.setZyyw(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_zyyw));
		investor.setJyyw(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_jyyw));
		investor.setIntro(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_intro));
		investor.setAccountant(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_accountant));
		investor.setLawyer(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_lawyer));
		investor.setSite(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_site));
		investor.setUrl(this.getRow(result, HbaseTableInvestorInfo.COLUMN_FAMILY, HbaseTableInvestorInfo.Column_Name_url));
		
		return investor;
	}
	
	/**
	 * 查询投资人财务信息
	 * @param investorID
	 * @param industryID
	 * @return
	 */
	public List<HBInvestorFinance> getInveFinance(int investorID, int industryID){
		List<HBInvestorFinance> ifList = new ArrayList<HBInvestorFinance>();
		this.currentClientNum++;
		Table table = null;
		
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorFinance.TABLE_NAME));
			String startRow = investorID + "_0";
			String stopRow = investorID + "_2100";
			Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
			ResultScanner rs = table.getScanner(scan);
			for(Result result: rs.next(100)){
				HBInvestorFinance inf = this.getFinanceEntity(result);
				String rowKey = Bytes.toString(result.getRow());
				inf.setYear(Integer.parseInt(rowKey.replace(investorID + "_", "")));
				inf.setInvestorID(investorID);
				inf.setIndustryID(industryID);
				ifList.add(inf);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return ifList;
	}
	
	/**
	 * 查询指定年份投资人财务信息
	 * @param investorID
	 * @param industryID
	 * @param year
	 * @return
	 */
	public HBInvestorFinance getInveFinanceByYear(int investorID, int industryID, int year){
		HBInvestorFinance inf = null;
		this.currentClientNum++;
		Table table = null;
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorFinance.TABLE_NAME));
			String rowKey = investorID + "_" + year;
			Get get = new Get(Bytes.toBytes(rowKey));
			Result result = table.get(get);
			if(result != null){
				inf = this.getFinanceEntity(result);
				inf.setIndustryID(industryID);
				inf.setInvestorID(investorID);
				inf.setYear(year);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return inf;
	}
	
	/**
	 * 将查询结果封装
	 * @param result
	 * @return
	 */
	protected HBInvestorFinance getFinanceEntity(Result result){
		HBInvestorFinance inf = new HBInvestorFinance();
		
		for (Cell cell : result.rawCells()) {
			String quilifier = Bytes.toString(CellUtil.cloneQualifier(cell));
			String value = Bytes.toString(CellUtil.cloneValue(cell));
			//只有一个family，因此不做family的判断
			if(quilifier.equals(HbaseTableInvestorFinance.Column_Name_asset)){inf.setAsset(value);}
			else if(quilifier.equals(HbaseTableInvestorFinance.Column_Name_currentAssets)){inf.setCurrentAssets(value);}
			else if(quilifier.equals(HbaseTableInvestorFinance.Column_Name_liabilities)){inf.setLiabilities(value);}
			else if(quilifier.equals(HbaseTableInvestorFinance.Column_Name_ownersEquity)){inf.setOwnersEquity(value);}
			else if(quilifier.equals(HbaseTableInvestorFinance.Column_Name_income)){inf.setIncome(value);}
			else if(quilifier.equals(HbaseTableInvestorFinance.Column_Name_cost)){inf.setCost(value);}
			else if(quilifier.equals(HbaseTableInvestorFinance.Column_Name_profit)){inf.setProfit(value);}
			else if(quilifier.equals(HbaseTableInvestorFinance.Column_Name_assetsGRate)){inf.setAssetsGRate(value);}
			else if(quilifier.equals(HbaseTableInvestorFinance.Column_Name_incomeGRate)){inf.setIncomeGRate(value);}
			else if(quilifier.equals(HbaseTableInvestorFinance.Column_Name_profitGRate)){inf.setProfitGRate(value);}
			else if(quilifier.equals(HbaseTableInvestorFinance.Column_Name_liabilitiesGRate)){inf.setLiabilitiesGRate(value);}
			else {
				this.logger.warn("unknown quilifier! quilifier: " + quilifier);
			}
		}
		
		return inf;
	}
	
	/**
	 * 查询投资人负责人信息
	 * @param investorID
	 * @param industryID
	 * @return
	 */
	public List<HBInvestorPrincipal> getInvePrincipals(int investorID, int industryID){
		List<HBInvestorPrincipal> ipList = new ArrayList<HBInvestorPrincipal>();
		this.currentClientNum++;
		Table table = null;
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorPrincipal.TABLE_NAME));
			String startRow = investorID + "_pri_0";
			String stopRow = investorID + "_pri_z";
			Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
			ResultScanner rs = table.getScanner(scan);
			for(Result result: rs.next(100)){
				HBInvestorPrincipal ip = this.getPrincipalEntiry(result);
				ip.setInvestorID(investorID);
				ip.setIndustryID(industryID);
				ipList.add(ip);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return ipList;
	}
	
	/**
	 * 将查询结果封装
	 * @param result
	 * @return
	 */
	protected HBInvestorPrincipal getPrincipalEntiry(Result result){
		HBInvestorPrincipal ip = new HBInvestorPrincipal();
		for (Cell cell : result.rawCells()) {
			String quilifier = Bytes.toString(CellUtil.cloneQualifier(cell));
			String value = Bytes.toString(CellUtil.cloneValue(cell));

			//只有一个family，因此不做family的判断
			if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_name)){ip.setName(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_job)){ip.setJob(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_info)){ip.setInfo(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_officePhone)){ip.setOfficePhone(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_mobilePhone)){ip.setMobilePhone(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_email)){ip.setEmail(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_qq)){ip.setQq(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_weixin)){ip.setWeixin(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_weibo)){ip.setWeibo(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_blog)){ip.setBlog(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_birthday)){ip.setBirthday(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_gender)){ip.setGender(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_rqksrq)){ip.setRqksrq(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_rqjsrq)){ip.setRqjsrq(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_degree)){ip.setDegree(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_gradSchool)){ip.setGradSchool(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_jobTitle)){ip.setJobTitle(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_party)){ip.setParty(value);}
			else if(quilifier.equals(HbaseTableInvestorPrincipal.Column_Name_workExper)){ip.setWorkExper(value);}
			else {
				this.logger.warn("unknown quilifier! quilifier: " + quilifier);;
			}
		}
		return ip;
	}
	
	/**
	 * 查询投资人产品信息
	 * @param investorID
	 * @param industryID
	 * @return
	 */
	public List<HBInvestorProduct> getInveProducts(int investorID, int industryID){
		List<HBInvestorProduct> ipList = new ArrayList<HBInvestorProduct>();
		this.currentClientNum++;
		Table table = null;
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorProduct.TABLE_NAME));
			String startRow = investorID + "_pro_0";
			String stopRow = investorID + "_pro_z";
			Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
			ResultScanner rs = table.getScanner(scan);
			for(Result result: rs.next(100)){
				HBInvestorProduct ip = this.getProductEntity(result);
				ip.setInvestorID(investorID);
				ip.setIndustryID(industryID);
				ipList.add(ip);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return ipList;
	}
	
	/**
	 * 将查询结果封装
	 * @param result
	 * @return
	 */	
	protected HBInvestorProduct getProductEntity(Result result){
		HBInvestorProduct ip = new HBInvestorProduct();
		for (Cell cell : result.rawCells()) {
			String quilifier = Bytes.toString(CellUtil.cloneQualifier(cell));
			String value = Bytes.toString(CellUtil.cloneValue(cell));

			//只有一个family，因此不做family的判断
			if(quilifier.equals(HbaseTableInvestorProduct.Column_Name_name)){ip.setName(value);}
			else if(quilifier.equals(HbaseTableInvestorProduct.Column_Name_type)){ip.setType(value);}
			else if(quilifier.equals(HbaseTableInvestorProduct.Column_Name_intro)){ip.setIntro(value);}
			else {
				this.logger.warn("unknown quilifier! quilifier: " + quilifier);;
			}
		}
		return ip;
	}
	
	/**
	 * 查询投资人资本动态
	 * @param investorID
	 * @param industryID
	 * @return
	 */
	public List<HBInvestorCapital> getInveCapitals(int investorID, int industryID){
		List<HBInvestorCapital> icList = new ArrayList<HBInvestorCapital>();
		this.currentClientNum++;
		Table table = null;
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorCapital.TABLE_NAME));
			String startRow = investorID + "_cap_0";
			String stopRow = investorID + "_cap_z";
			Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
			ResultScanner rs = table.getScanner(scan);
			for(Result result: rs.next(100)){
				HBInvestorCapital ip = this.getCapitalEntity(result);
				ip.setInvestorID(investorID);
				ip.setIndustryID(industryID);
				icList.add(ip);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return icList;
	}
	
	/**
	 * 将查询结果封装
	 * @param result
	 * @return
	 */	
	protected HBInvestorCapital getCapitalEntity(Result result){
		HBInvestorCapital ic = new HBInvestorCapital();
		for (Cell cell : result.rawCells()) {
			String quilifier = Bytes.toString(CellUtil.cloneQualifier(cell));
			String value = Bytes.toString(CellUtil.cloneValue(cell));

			//只有一个family，因此不做family的判断
			if(quilifier.equals(HbaseTableInvestorCapital.Column_Name_rzxq)){ic.setRzxq(value);}
			else if(quilifier.equals(HbaseTableInvestorCapital.Column_Name_rzed)){ic.setRzed(value);}
			else if(quilifier.equals(HbaseTableInvestorCapital.Column_Name_jhsssj)){ic.setJhsssj(value);}
			else if(quilifier.equals(HbaseTableInvestorCapital.Column_Name_jnssbk)){ic.setJnssbk(value);}
			else if(quilifier.equals(HbaseTableInvestorCapital.Column_Name_jwssbk)){ic.setJwssbk(value);}
			else if(quilifier.equals(HbaseTableInvestorCapital.Column_Name_sjscjd)){ic.setSjscjd(value);}
			else if(quilifier.equals(HbaseTableInvestorCapital.Column_Name_bgxq)){ic.setBgxq(value);}
			else {
				this.logger.warn("unknown quilifier! quilifier: " + quilifier);;
			}
		}
		return ic;
	}
	
	/**
	 * 查询投资人社会关系
	 * @param investorID
	 * @param industryID
	 * @return
	 */
	public List<HBInvestorSocialRelations> getInveSocial(int investorID, int industryID){
		List<HBInvestorSocialRelations> isrList = new ArrayList<HBInvestorSocialRelations>();
		this.currentClientNum++;
		Table table = null;
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorSocialRelations.TABLE_NAME));
			String startRow = investorID + "_soc_0";
			String stopRow = investorID + "_soc_z";
			Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
			ResultScanner rs = table.getScanner(scan);
			for(Result result: rs.next(100)){
				HBInvestorSocialRelations ip = this.getSocialEntity(result);
				ip.setInvestorID(investorID);
				ip.setIndustryID(industryID);
				isrList.add(ip);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return isrList;
	}
	
	/**
	 * 将查询结果封装
	 * @param result
	 * @return
	 */	
	protected HBInvestorSocialRelations getSocialEntity(Result result){
		HBInvestorSocialRelations isr = new HBInvestorSocialRelations();
		for (Cell cell : result.rawCells()) {
			String quilifier = Bytes.toString(CellUtil.cloneQualifier(cell));
			String value = Bytes.toString(CellUtil.cloneValue(cell));

			//只有一个family，因此不做family的判断
			if(quilifier.equals(HbaseTableInvestorSocialRelations.Column_Name_name)){isr.setName(value);}
			else if(quilifier.equals(HbaseTableInvestorSocialRelations.Column_Name_type)){isr.setType(value);}
			else if(quilifier.equals(HbaseTableInvestorSocialRelations.Column_Name_information)){isr.setInformation(value);}
			else if(quilifier.equals(HbaseTableInvestorSocialRelations.Column_Name_company)){isr.setCompany(value);}
			else {
				this.logger.warn("unknown quilifier! quilifier: " + quilifier);;
			}
		}
		return isr;
	}
	
	/**
	 * 插入投资人基本信息
	 * @param investor
	 * @return status
	 */
	public HbaseStatus insertInvestor(HBInvestorInfo investor){
		HbaseStatus status = HbaseStatus.successed;
		this.currentClientNum++;
		Table table = null;
		if(investor.getInvestorID() <= 0){
			return HbaseStatus.illogic;
		}
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorInfo.TABLE_NAME));
			String rowKey = "" + investor.getInvestorID();
			Put put = new Put(Bytes.toBytes(rowKey));
			byte[] famByt = Bytes.toBytes(HbaseTableInvestorInfo.COLUMN_FAMILY);
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_name, investor.getName());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_industry, investor.getIndustry());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_province, investor.getProvince());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_city, investor.getCity());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_county, investor.getCounty());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_abbreviation, investor.getAbbreviation());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_enName, investor.getEnName());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_parentCompany, investor.getParentCompany());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_registeredNo, investor.getRegisteredNo());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_legalPerson, investor.getLegalPerson());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_regDate, investor.getRegDate());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_dateRange, investor.getDateRange());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_orgCode, investor.getOrgCode());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_isIPO, investor.getIsIPO());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_stockCode, investor.getStockCode());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_property, investor.getProperty());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_regCapital, investor.getRegCapital());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_address, investor.getAddress());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_website, investor.getWebsite());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_weixin, investor.getWeixin());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_weibo, investor.getWeibo());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_telephone, investor.getTelephone());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_email, investor.getEmail());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_fax, investor.getFax());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_businessScope, investor.getBusinessScope());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_contackAdd, investor.getContackAdd());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_orgForm, investor.getOrgForm());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_officeAdd, investor.getOfficeAdd());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_employees, investor.getEmployees());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_zyyw, investor.getZyyw());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_jyyw, investor.getJyyw());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_intro, investor.getIntro());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_accountant, investor.getAccountant());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_lawyer, investor.getLawyer());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_site, investor.getSite());
			this.addValue(put, famByt, HbaseTableInvestorInfo.Column_Name_url, investor.getUrl());
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return status;
	}
	
	/**
	 * 插入投资人资本动态
	 * @param capital
	 * @return status
	 */
	public HbaseStatus insertCapital(HBInvestorCapital capital){
		HbaseStatus status = HbaseStatus.successed;
		this.currentClientNum++;
		Table table = null;
		if(capital.getInvestorID() <= 0){
			return HbaseStatus.illogic;
		}
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorCapital.TABLE_NAME));
			String rowKey = capital.getInvestorID() + "_" + Calendar.getInstance().getTimeInMillis();
			Put put = new Put(Bytes.toBytes(rowKey));
			byte[] famByt = Bytes.toBytes(HbaseTableInvestorCapital.COLUMN_FAMILY);
			this.addValue(put, famByt, HbaseTableInvestorCapital.Column_Name_rzxq, capital.getRzxq());
			this.addValue(put, famByt, HbaseTableInvestorCapital.Column_Name_rzed, capital.getRzed());
			this.addValue(put, famByt, HbaseTableInvestorCapital.Column_Name_jhsssj, capital.getJhsssj());
			this.addValue(put, famByt, HbaseTableInvestorCapital.Column_Name_jnssbk, capital.getJnssbk());
			this.addValue(put, famByt, HbaseTableInvestorCapital.Column_Name_jwssbk, capital.getJwssbk());
			this.addValue(put, famByt, HbaseTableInvestorCapital.Column_Name_sjscjd, capital.getSjscjd());
			this.addValue(put, famByt, HbaseTableInvestorCapital.Column_Name_bgxq, capital.getBgxq());
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return status;
	}
	
	/**
	 * 插入投资人财务数据
	 * @param capital
	 * @return status
	 */
	public HbaseStatus insertFinance(HBInvestorFinance finance){
		HbaseStatus status = HbaseStatus.successed;
		this.currentClientNum++;
		Table table = null;
		if(finance.getInvestorID() <= 0 || finance.getYear() <= 1900){
			return HbaseStatus.illogic;
		}
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorFinance.TABLE_NAME));
			String rowKey = finance.getInvestorID() + "_" + finance.getYear();
			Put put = new Put(Bytes.toBytes(rowKey));
			byte[] famByt = Bytes.toBytes(HbaseTableInvestorFinance.COLUMN_FAMILY);
			this.addValue(put, famByt, HbaseTableInvestorFinance.Column_Name_asset, finance.getAsset());
			this.addValue(put, famByt, HbaseTableInvestorFinance.Column_Name_currentAssets, finance.getCurrentAssets());
			this.addValue(put, famByt, HbaseTableInvestorFinance.Column_Name_liabilities, finance.getLiabilities());
			this.addValue(put, famByt, HbaseTableInvestorFinance.Column_Name_ownersEquity, finance.getOwnersEquity());
			this.addValue(put, famByt, HbaseTableInvestorFinance.Column_Name_income, finance.getIncome());
			this.addValue(put, famByt, HbaseTableInvestorFinance.Column_Name_cost, finance.getCost());
			this.addValue(put, famByt, HbaseTableInvestorFinance.Column_Name_profit, finance.getProfit());
			this.addValue(put, famByt, HbaseTableInvestorFinance.Column_Name_assetsGRate, finance.getAssetsGRate());
			this.addValue(put, famByt, HbaseTableInvestorFinance.Column_Name_incomeGRate, finance.getIncomeGRate());
			this.addValue(put, famByt, HbaseTableInvestorFinance.Column_Name_profitGRate, finance.getProfitGRate());
			this.addValue(put, famByt, HbaseTableInvestorFinance.Column_Name_liabilitiesGRate, finance.getLiabilitiesGRate());
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return status;
	}
	
	/**
	 * 插入投资人负责人
	 * @param capital
	 * @return status
	 */
	public HbaseStatus insertPrincipal(HBInvestorPrincipal principal){
		HbaseStatus status = HbaseStatus.successed;
		this.currentClientNum++;
		Table table = null;
		if(principal.getInvestorID() <= 0 || principal.getName() == null || principal.getName().length() == 0){
			return HbaseStatus.illogic;
		}
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorPrincipal.TABLE_NAME));
			String rowKey = principal.getInvestorID() + "_pri_" + MD5Util.createMd5(principal.getName());
			Put put = new Put(Bytes.toBytes(rowKey));
			byte[] famByt = Bytes.toBytes(HbaseTableInvestorPrincipal.COLUMN_FAMILY);
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_name, principal.getName());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_job, principal.getJob());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_info, principal.getInfo());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_officePhone, principal.getOfficePhone());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_mobilePhone, principal.getMobilePhone());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_email, principal.getEmail());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_qq, principal.getQq());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_weixin, principal.getWeixin());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_weibo, principal.getWeibo());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_blog, principal.getBlog());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_birthday, principal.getBirthday());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_gender, principal.getGender());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_rqksrq, principal.getRqksrq());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_rqjsrq, principal.getRqjsrq());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_degree, principal.getDegree());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_gradSchool, principal.getGradSchool());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_jobTitle, principal.getJobTitle());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_party, principal.getParty());
			this.addValue(put, famByt, HbaseTableInvestorPrincipal.Column_Name_workExper, principal.getWorkExper());

			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return status;
	}
	
	/**
	 * 插入投资人产品信息
	 * @param capital
	 * @return status
	 */
	public HbaseStatus insertProduct(HBInvestorProduct product){
		HbaseStatus status = HbaseStatus.successed;
		this.currentClientNum++;
		Table table = null;
		if(product.getInvestorID() <= 0 || product.getName() == null || product.getName().length() == 0){
			return HbaseStatus.illogic;
		}
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorProduct.TABLE_NAME));
			String rowKey = product.getInvestorID() + "_pro_" + MD5Util.createMd5(product.getName());
			Put put = new Put(Bytes.toBytes(rowKey));
			byte[] famByt = Bytes.toBytes(HbaseTableInvestorProduct.COLUMN_FAMILY);
			this.addValue(put, famByt, HbaseTableInvestorProduct.Column_Name_name, product.getName());
			this.addValue(put, famByt, HbaseTableInvestorProduct.Column_Name_type, product.getType());
			this.addValue(put, famByt, HbaseTableInvestorProduct.Column_Name_intro, product.getIntro());
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return status;
	}
	
	/**
	 * 插入投资人社会关系
	 * @param capital
	 * @return status
	 */
	public HbaseStatus insertSocial(HBInvestorSocialRelations social){
		HbaseStatus status = HbaseStatus.successed;
		this.currentClientNum++;
		Table table = null;
		if(social.getInvestorID() <= 0 || social.getName() == null || social.getName().length() == 0){
			return HbaseStatus.illogic;
		}
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableInvestorSocialRelations.TABLE_NAME));
			String rowKey = social.getInvestorID() + "_soc_" + MD5Util.createMd5(social.getName());
			Put put = new Put(Bytes.toBytes(rowKey));
			byte[] famByt = Bytes.toBytes(HbaseTableInvestorSocialRelations.COLUMN_FAMILY);
			this.addValue(put, famByt, HbaseTableInvestorSocialRelations.Column_Name_name, social.getName());
			this.addValue(put, famByt, HbaseTableInvestorSocialRelations.Column_Name_type, social.getType());
			this.addValue(put, famByt, HbaseTableInvestorSocialRelations.Column_Name_information, social.getInformation());
			this.addValue(put, famByt, HbaseTableInvestorSocialRelations.Column_Name_company, social.getCompany());
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return status;
	}
	
	/**
	 * 赋值
	 * @param put
	 * @param famByt
	 * @param qualifier
	 * @param value
	 */
	private void addValue(Put put, byte[] famByt, String qualifier, String value){
		if(value != null){
			put.addColumn(famByt, Bytes.toBytes(qualifier), Bytes.toBytes(value));
		}
	}
	
	/**
	 * 获取指定列的值
	 * @param result
	 * @param familly
	 * @param rowName
	 * @return
	 */
	protected String getRow(Result result,String familly,String rowName){
		byte[] byts = result.getValue(Bytes.toBytes(familly), Bytes.toBytes(rowName));
		String row = null;
		if (byts != null && byts.length > 0) {
			row = Bytes.toString(byts);
		}
		return row;
	}

	private void testConn(){
		if(this.conn == null){
			this.initConn();
		}
	}

	/*protected void close(ResultScanner rs, Table table) {
		try {
			rs.close();
		} catch (Exception e) {}
		try {
			table.close();
		} catch (Exception e1) {}
	}*/
	
	/*private void closeConn(){
		try {
			this.conn.close();
		} catch (Exception e) {}
		this.conn = null;
	}*/
	private void closeConn(Table table,Connection conn){
		try {
			table.close();
			conn.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		
	}
	/*public void closeClient(){
		this.closeConn();
		try {
			this.timer.cancel();
		} catch (Exception e) {}
	}*/
}
